package com.xinqing.bigdata.flink.datastream.demo;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * @Author:CHQ
 * @Date:2021/4/19 16:27
 * @Description
 */
public class WC {
    public static void main(String args[]) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //按照顺序将元素flatmap打散并处理输出一个二元组(k,1)字样，后按key归类，最后sum二元组的第二个位置元素
        SingleOutputStreamOperator<Tuple2<String, Integer>> source = env.socketTextStream("localhost", 9000) //cmd: nc -l -p9000
                .flatMap(new MyFilter()) //faltmap跟map的区别在于，flatmap可将条记录分开处理并输出，而map意在与将整个条记录看做一个处理单元
                .keyBy(tuple2 -> tuple2.f0) //将二元组的第一个元素作为分类key
                //分类后开一个5s的窗，统计输出。其实此处也可以放到flatmap后，
                // keyby之前(即意为在打算记录并输出后，立马开窗统计输出，此时的开窗针对的是所有的key)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .sum(1);

        source.print();

        env.execute();
    }


    private static class MyFilter implements org.apache.flink.api.common.functions.FlatMapFunction<String, Tuple2<String, Integer>> {

        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {

            for (String s : value.split(" ")) {
                out.collect(new Tuple2<>(s, 1));
            }

        }
    }
}
